# coding:utf8
"""
This is Code Description:GenderModer
"""
from pyspark import SparkContext
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import LongType

from index_development.BaseModel.BaseModelAbstract import BaseModel


class GenderModel(BaseModel):
    def getTagId(self):
        return 7

    def compute(self, esDF: DataFrame, fiveDF: DataFrame, sc: SparkSession, spark: SparkContext):
    #     print("=================01-esDF================================")
    #     esDF.show()
    #     esDF.printSchema()
        # +---+-------+
        # | sex | user_id |
        # +---+-------+
        # | M | 22 - 540 |
        # | M | 292 - 162 |

        # print("=================02-fiveDF================================")
        # fiveDF.show()
        # fiveDF.printSchema()
        # +---+----+
        # | id | rule |
        # +---+----+
        # | 14 | 男 |
        # | 15 | 女 |
        # +---+----+

        #TODO 6.将esDF的sex中的'M→男,F→女
        print("=================03-：将esDF的sex中的'M→男,F→女'================================")
        # 方法1：
        from pyspark.sql.functions import regexp_replace
        esDF2 = esDF.select(regexp_replace(regexp_replace(esDF['sex'], 'M', '男'), 'F', '女').alias('sex'),
                            esDF["user_id"].alias("user_id"))
        esDF2.show()
        esDF2.printSchema()
        # +---+-------+
        # | sex | user_id |
        # +---+-------+
        # | 男 | 22 - 540 |
        # | 男 | 292 - 162 |
        # | 男 | 332 - 107 |
        # | 男 | 66 - 399 |
        # | 女 | 69 - 108 |
        # 方法2分两步：
        # from pyspark.sql.functions import regexp_replace
        # print("=================03.1 - 将F转成'女'================================")
        # esDF2 = esDF.select(esDF["user_id"].alias("id"),regexp_replace(esDF["sex"], "F", "女").alias("sex"))
        # esDF2.show()
        # print("=================03.2 - 将M转成'男'================================")
        # esDF3 = esDF2.select(esDF2["id"],regexp_replace(esDF2["sex"], "M", "男").alias("sex"))
        # esDF3.show()

        # TODO 7.将fiveDF转换成fiveDict字典
        print("=================04.将fiveDF转换成fiveDict字典'================================")
        fiveDict: dict = fiveDF.rdd.map(lambda row: (row["rule"], row["id"])).collectAsMap()
        print(fiveDict)
        print(type(fiveDict))
        # {'男': 14, '女': 15}

        #TODO 8.获得newDF
        @udf
        def SexModelTotag(sex: str):
            return fiveDict[str(sex)]
        newDF:DataFrame = esDF2.select(esDF2["user_id"].alias("userId"),SexModelTotag(esDF2["sex"]).alias("tagsId"))
        newDF.show()
        newDF.printSchema()
        return newDF
        # | userId|tagsId|
        # +-------+------+
        # | 22-540|    14|
        # |292-162|    14|


if __name__ == '__main__':
    gender_mode = GenderModel()
    gender_mode.execute()
